package ckR

//
//import (
//	"fmt"
//	"db2s/global"
//	"db2s/scheduleTask"
//	mq "db2s/topic-mq"
//	"runtime"
//	"strings"
//	"sync"
//	"sync/atomic"
//	"time"
//)

//// rowsIsSegmentQueryListeners IsSegmentQuery  监听执行器
//func rowsIsSegmentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		logSeq        = message.Properties["logSeq"].(int64)
//		dTopic        = IsSegmentQuery
//		event         = "[rowsIsSegmentQueryListeners]"
//		wg            sync.WaitGroup
//		done          = make(chan struct{})
//		done1         = make(chan struct{}, 1)
//		mqTaskSum     int64
//		deleteMapInfo = map[string]any{"logSeq": logSeq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	pc, _, _, _ := runtime.Caller(1)
//	callingFunc := runtime.FuncForPC(pc)
//	callFile, callLine := callingFunc.FileLine(pc)
//	callFunc := callingFunc.Name()
//	callFuncInfo := fmt.Sprintf("Call functions %v Call File %v:%v", callFunc, callFile, callLine)
//	plan.Status.SubTaskBeginTime = time.Now().UnixNano()
//	if GlobalPConfigs.result.Teletypewriter == "bar" {
//		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "running")
//		global.Wlog.Debug(fmt.Sprintf("(%d) %v %s Add task Bar Run object %v successfully", logSeq, callFuncInfo, event, tableName))
//	}
//	InitDataFixFile(plan)
//	//处理wheresql
//	if v, ok := GlobalPConfigs.schema.WhereSql[fmt.Sprintf("%s%s%s", plan.Object.Schema, global.SchemaTableSplit, plan.Object.Table)]; ok && v != "" {
//		mqTaskSum = 1
//		plan.whereSqlFilter(&wg, message, v)
//	} else {
//		mqTaskSum = int64(len(plan.Object.PartitionData.Name.Source))
//		plan.IsSegmentSendMessage(&wg, message)
//	}
//	wg.Add(3)
//	go plan.speedMonitor(done1, &wg)
//	//init status
//	go expensiveOperation(&wg, message, done, &mqTaskSum, logSeq)
//	go mqTableStatusMonitor(done, done1, &wg, plan, logSeq)
//	// 获取当前计数
//	wg.Wait()
//	return mq.ConsumeSuccess
//}
//
//// rowsIndexConcurrentQueryListeners IndexConcurrentQuery 监听执行器
//func rowsIndexConcurrentQueryListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		topic         = QueryTableSql
//		dTopic        = IndexConcurrentQuery
//		event         = "[rowsIndexConcurrentQueryListeners]"
//		fatherSonTags = make(map[string][]string)
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter} //
//	)
//	defer deleteTopic(deleteMapInfo)
//	segmentSql := message.Properties["segmentSql"].(global.StartPart)
//	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 || len(segmentSql.Error) > 0 {
//		plan.errorAction(message, logseq, segmentSql.Error)
//		return mq.ConsumeSuccess
//	}
//	//分段查询错误
//	taskObject := message.Properties["taskObject"].(scheduleTask.TaskDistribution)
//	var l = chanSendMsg{
//		Object:   "second",
//		SendData: taskObject.IsConcurrentValue(segmentSql),
//		SendMsgInfo: SendMsgInfo{
//			CheckMod:  getCheckMod(),
//			Topic:     topic,
//			TableName: tableName,
//		},
//		SendMsg: sendMsg,
//		FatherSon: FatherSon{
//			FatherTags: map[string]any{"logSeq": logseq,
//				"taskObject": message.Properties["taskObject"],
//				"sendMsgKey": "sqlWhere", "subTask": plan.subTaskInfo, "label": message.Properties["label"],
//				"plan": message.Properties["plan"], "sourcePartitionName": message.Properties["sourcePartitionName"],
//				"insertPrefix": message.Properties["insertPrefix"],
//				"topic":        string(topic), "outFay": GlobalPConfigs.result.Teletypewriter},
//			FatherSonTags: fatherSonTags,
//		},
//		SendMsgEnd: func() {
//		},
//	}
//	if !chanMonitorSendMsg(&l) {
//		return mq.ReconsumeLater
//	}
//	if atomic.LoadInt64(plan.Status.ErrEnd) > -1 {
//		plan.mqProductCustomerMonitor.ProductMessage <- fatherSonTags
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqLevelCustomerReceive, 1)
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsQueryTableSqlListeners rowsQueryTableSql 监听执行器
//func rowsQueryTableSqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		logseq        = message.Properties["logSeq"].(int64)
//		topic         = QueryTableData
//		dTopic        = QueryTableSql
//		event         = "[rowsQueryTableSqlListeners]"
//		tableSql      querySql
//		err           error
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	sqlWhere := message.Properties["sqlWhere"].(global.StartPart)
//	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 || len(sqlWhere.Error) > 0 {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		plan.errorAction(message, logseq, sqlWhere.Error)
//		return mq.ReconsumeLater
//	}
//	partitionName := message.Properties["sourcePartitionName"].(string)
//	if tableSql, err = plan.QuerySql(sqlWhere, partitionName); err != nil {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"taskObject": message.Properties["taskObject"], "label": message.Properties["label"],
//		"strSql": tableSql, "err": plan.Status.ErrEnd, "sourcePartitionName": message.Properties["sourcePartitionName"],
//		"insertPrefix": message.Properties["insertPrefix"],
//		"subTask":      plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsQueryTableDataSqlListeners QueryTableData 监听执行器
//func rowsQueryTableDataSqlListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		logSeq        = message.Properties["logSeq"].(int64)
//		tableName     = string(message.Body)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		sqld          sqlData
//		err           error
//		event         = "[rowsQueryTableDataSqlListeners]"
//		topic         = Md5Rows
//		dTopic        = QueryTableData
//		deleteMapInfo = map[string]any{"logSeq": logSeq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	defer func() {
//		atomic.AddInt64(plan.TableSpeedLimit.ReadParallel, -1)
//		atomic.AddInt64(plan.TableSpeedLimit.WriteParallel, -1)
//	}()
//	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		return mq.ConsumeSuccess
//	}
//	tableSql := message.Properties["strSql"].(querySql)
//	partitionName := message.Properties["sourcePartitionName"].(string)
//	atomic.AddInt64(plan.TableSpeedLimit.ReadParallel, 1)
//	atomic.AddInt64(plan.TableSpeedLimit.WriteParallel, 1)
//	if sqld, err = plan.QueryTableData(tableSql, partitionName, logSeq); err != nil {
//		return 0
//	}
//	if GlobalPConfigs.result.Teletypewriter == "bar" {
//		if PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) != -2 {
//			PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": plan.subTaskInfo, "select": sqld.Ssum, "insert": sqld.Dsum})
//		}
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logSeq, "plan": message.Properties["plan"],
//		"label": message.Properties["label"], "insertPrefix": message.Properties["insertPrefix"],
//		"sourceD": sqld.Source, "destD": sqld.Dest, "err": plan.Status.ErrEnd, "subTask": plan.subTaskInfo, "sourcePartitionName": message.Properties["sourcePartitionName"],
//		"topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsMd5Listeners rowsMd5 监听执行器
//func rowsMd5Listeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logSeq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		topic         = Arrcmp
//		dTopic        = Md5Rows
//		event         = "[rowsMd5Listeners]"
//		deleteMapInfo = map[string]any{"logSeq": logSeq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	stt, dtt := message.Properties["sourceD"], message.Properties["destD"]
//	stt1, dtt1, abs := plan.Md5Rows(stt, dtt, logSeq)
//	if !abs {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logSeq, "plan": message.Properties["plan"],
//		"label": message.Properties["label"], "sourceD": stt1, "destD": dtt1,
//		"insertPrefix": message.Properties["insertPrefix"], "sourcePartitionName": message.Properties["sourcePartitionName"],
//		"subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsArrcmpListeners rowsArrcmp 监听执行器
//func rowsArrcmpListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logSeq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		topic         = AbnormalDataDiff
//		dTopic        = Arrcmp
//		event         = "[rowsArrcmpListeners]"
//		deleteMapInfo = map[string]any{"logSeq": logSeq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	stt, dtt := message.Properties["sourceD"], message.Properties["destD"]
//	dataDisPos := plan.Arrcmp(stt, dtt, logSeq)
//	if len(dataDisPos) == 0 {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		return mq.ConsumeSuccess
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logSeq, "plan": message.Properties["plan"],
//		"label":        message.Properties["label"],
//		"insertPrefix": message.Properties["insertPrefix"], "sourcePartitionName": message.Properties["sourcePartitionName"],
//		"dataDisPos": dataDisPos, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsAbnormalDataDiffListeners AbnormalDataDiff 监听执行器
//func rowsAbnormalDataDiffListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		topic         = DataFixDiff
//		dTopic        = AbnormalDataDiff
//		event         = "[rowsAbnormalDataDiffListeners]"
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	insertPrefix := message.Properties["insertPrefix"].(string)
//	dataDisPos := message.Properties["dataDisPos"].(map[string]any)
//	diff, err := plan.AbnormalDataDiff(dataDisPos, insertPrefix, logseq)
//	if err != nil {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		plan.errorAction(message, logseq, err)
//		return mq.ConsumeSuccess
//	}
//	if diff.Insert == nil && diff.Delete == nil {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//		return mq.ReconsumeLater
//	}
//	if !sendMsg(getCheckMod(), topic, tableName, map[string]any{"logSeq": logseq, "plan": message.Properties["plan"],
//		"label": message.Properties["label"], "sourcePartitionName": message.Properties["sourcePartitionName"],
//		"insertPrefix": message.Properties["insertPrefix"],
//		"diffSql":      diff, "subTask": plan.subTaskInfo, "topic": string(topic), "outFay": GlobalPConfigs.result.Teletypewriter}) {
//		return mq.ReconsumeLater
//	}
//	return mq.ConsumeSuccess
//}
//
//// rowsDataFixDiffListeners DataFixDiff 监听执行器
//func rowsDataFixDiffListeners(message mq.Message) mq.ConsumeConcurrentlyStatus {
//	var (
//		//tableName     = string(message.Body)
//		logseq        = message.Properties["logSeq"].(int64)
//		plan          = message.Properties["plan"].(*SchedulePlan2)
//		dTopic        = DataFixDiff
//		event         string
//		partitionName = message.Properties["sourcePartitionName"].(string)
//		insertPrefix  = message.Properties["insertPrefix"].(string)
//		deleteMapInfo = map[string]any{"logSeq": logseq, "message": message, "event": event, "dtopic": dTopic, "subTask": plan.subTaskInfo, "topic": string(dTopic), "outFay": GlobalPConfigs.result.Teletypewriter}
//	)
//	defer deleteTopic(deleteMapInfo)
//	defer func() {
//		atomic.AddInt64(plan.mqProductCustomerMonitor.MqCustomerSeq, 1)
//	}()
//	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
//		return mq.ConsumeSuccess
//	}
//	diffSql := message.Properties["diffSql"]
//	if strings.EqualFold(GlobalPConfigs.repair.DataFix, "no") {
//		return mq.ConsumeSuccess
//	}
//	if err := plan.writeFixSql(false, insertPrefix, diffSql.(FixSql), partitionName, logseq); err != nil {
//		plan.errorAction(message, logseq, err)
//		return mq.ConsumeSuccess
//	}
//	return mq.ConsumeSuccess
//}
